package org.wisdom.mq.kafka.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
import org.wisdom.mq.kafka.service.KafkaConsumer;

/**
 * 生产消息
 *
 * @author 张智研
 * @date 2023/11/09 09:32
 **/
@RestController
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private KafkaConsumer kafkaConsumer;

    @GetMapping("/kafka/normal/{message}")
    public void sendNormalMessage(@PathVariable("message") String message) {

        kafkaTemplate.send("sb_topic", message).addCallback(success -> {
            // 消息发送到的topic
            assert success != null;
            String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            System.out.println("发送消息成功1:" + topic + "-" + partition + "-" + offset);
        }, throwable -> System.out.println("发送消息失败1:" + throwable.getMessage()));
    }


    @PostMapping("/kafka/normal2")
    public void sendNormalMessage22(String sss) {
        System.out.println(sss);
    }
}
